/*
 *  ------------------------------------------------------------------
 *  Copyright © 2017 Hangzhou DtDream Technology Co.,Lt d. All rights reserved.
 *  ------------------------------------------------------------------
 *              Product: EMR
 *    Module Name: DataTrain
 *      Date Created: 2017-10-23
 *         Description:
 *  ------------------------------------------------------------------
 *  Modification History
 *  DATE            Name           Description
 *  ------------------------------------------------------------------
 *  2017-10-23     XZ 209
 *  ------------------------------------------------------------------
 */

package org.xukai.remoting.sdk.rpc.transfer;

import com.dtdream.emr.transmission.exception.TransmissionException;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.util.Arrays;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Queue;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;

import static com.dtdream.emr.transmission.exception.TransmissionException.Error.EXCEPTION_SDK_CONFIG;
import static com.dtdream.emr.transmission.exception.TransmissionException.Error.EXCEPTION_SDK_NETWORK;

public class TransferNodePool implements AutoCloseable {
    private static final Logger LOG = LoggerFactory.getLogger(TransferNodePool.class);

    private static final AtomicInteger sign = new AtomicInteger(new Random().nextInt());

    /* 用户认证信息 */
    private String accessKey;
    private String keySecret;

    private EventLoopGroup workerGroup;
    /* 和单个节点的通道信息 */
    private List<TransferNode> nodes = new LinkedList<TransferNode>();
    private int index = 0;

    public TransferNodePool(String accessKey, String keySecret) {
        this.accessKey = accessKey;
        this.keySecret = keySecret;
    }

    /**
     * 初始化的时候需要连接集群所有节点，获取数据通道端口号
     *
     * @param endpoints
     */
    public void init(String endpoints, NettyConfig nettyConfig) throws TransmissionException {
        String[] urls = endpoints.split(",");
        if (urls.length <= 0) {
            throw new TransmissionException(EXCEPTION_SDK_CONFIG, "No valid server urls");
        }
        //init netty thread pool
        this.workerGroup = new NioEventLoopGroup(nettyConfig.getEventLoopThreads());

        Queue<InetSocketAddress> unseenAddresses = parseAddresses(Arrays.asList(urls));
        Set<InetSocketAddress> seenAddresses = new HashSet<>();
        while (!unseenAddresses.isEmpty()) {
            try {
                TransferNode node = new TransferNode(
                        unseenAddresses.poll(),
                        accessKey,
                        keySecret,
                        workerGroup,
                        nettyConfig);
                nodes.add(node);
                node.init();
                seenAddresses.add(node.address());
                for (InetSocketAddress address : parseAddresses(node.controlServers())) {
                    if (!seenAddresses.contains(address)) {
                        unseenAddresses.offer(address);
                    }
                }
            } catch (TransmissionException | IOException e) {
                LOG.warn("Init TransferNode error: " + e.getMessage());
            }
        }
        if (nodes.isEmpty()) {
            throw new TransmissionException(EXCEPTION_SDK_NETWORK, "No server available");
        } else {
            LOG.info("TransferNodePool inited with nodes: " + seenAddresses);
        }
        this.index = Math.abs(sign.incrementAndGet() % nodes.size());
    }

    static LinkedList<InetSocketAddress> parseAddresses(List<String> addresses) {
        LinkedList<InetSocketAddress> validAddresses = new LinkedList<>();
        for (String address : addresses) {
            int splitPoint = address.indexOf(":");
            String errorMessage = "Invalid server address: " + address;

            InetSocketAddress socketAddress = null;
            if (splitPoint > -1) {
                String host = address.substring(0, splitPoint);
                try (Socket socket = new Socket()) {
                    Integer port = Integer.valueOf(address.substring(splitPoint + 1));
                    socketAddress = new InetSocketAddress(host, port);
                    socket.connect(socketAddress, 5000);
                    validAddresses.add(socketAddress);
                } catch (NumberFormatException e) {
                    LOG.warn(errorMessage);
                } catch (IOException e) {
                    LOG.warn("Server is unreachable: " + address);
                }
            } else {
                LOG.warn(errorMessage);
            }
        }
        return validAddresses;
    }

    /**
     * 获取和节点的连接，使用场景：
     * 1. 创建Writer时，指定节点发送，后续此Writer数据都走该通道
     *
     * @return
     */
    public synchronized TransferNode next() {
        index %= nodes.size();
        TransferNode ret = nodes.get(index);
        index++;
        return ret;
    }

    public int size() {
        return nodes.size();
    }

    public void close() {
        for (TransferNode transferNode : nodes) {
            transferNode.close();
        }
        if (null != workerGroup) {
            workerGroup.shutdownGracefully();
            LOG.info("Release eventloop group succeed! ");
        }
    }

}
